package connect

import (
	"bytes"
	"encoding/json"
	"litetalk/config"
	"litetalk/proto"
	"net"
	"strings"
	"time"

	"github.com/sirupsen/logrus"
)

func GoTcp() {
	DefaultServer.Buckets = make([]*Bucket, config.Conf.Connect.CpuNum)
	DefaultServer.BucketIdx = len(DefaultServer.Buckets)
	for i := 0; i < DefaultServer.BucketIdx; i++ {
		DefaultServer.Buckets[i] = NewBucket()
	}
	InitTcpServer()
}

func InitTcpServer() error {
	tcpAddr := strings.Split(config.Conf.Tcp.Bind, ",")

	var (
		addr     *net.TCPAddr
		listener *net.TCPListener
		err      error
	)

	for _, ipPort := range tcpAddr {
		if addr, err = net.ResolveTCPAddr("tcp", ipPort); err != nil {
			return err
		}
		if listener, err = net.ListenTCP("tcp", addr); err != nil {
			return err
		}
		logrus.Infof("tcp listen on addr: %s", addr)

		go AcceptTcp(listener)
	}
	return nil
}

func AcceptTcp(listener *net.TCPListener) {
	var (
		conn *net.TCPConn
		err  error
	)

	config := config.Conf.Tcp
	for {
		if conn, err = listener.AcceptTCP(); err != nil {
			logrus.Errorf("listener.Accept(\"%s\") error(%v)", listener.Addr().String(), err)
			return
		}
		if err = conn.SetKeepAlive(config.KeepAlive); err != nil {
			logrus.Errorf("conn.SetKeepAlive() error:%s", err.Error())
			return
		}
		if err = conn.SetReadBuffer(config.ReceiveBuf); err != nil {
			logrus.Errorf("conn.SetReadBuffer() error:%s", err.Error())
			return
		}
		if err = conn.SetWriteBuffer(config.SendBuf); err != nil {
			logrus.Errorf("conn.SetWriteBuffer() error:%s", err.Error())
			return
		}
		go ServeTcp(&DefaultServer, conn)
	}
}

func ServeTcp(s *Server, conn *net.TCPConn) {
	ch := NewChannel(s.BroadcastSize)
	ch.connTcp = conn

	go readFromTcp(s, ch)
	go writeFromTcp(s, ch)
}

func readFromTcp(s *Server, ch *Channel) {
	defer func() {
		logrus.Infof("start exec disconnect ...")
		s.Bucket(ch.userId).DeleteChannel(ch)

		disconnectReq := &proto.DisConnect{
			AuthToken: ch.authToken,
		}
		rep := &proto.DisConnectReply{}
		err := logicRpc.DisConnect(disconnectReq, rep)
		if err != nil {
			logrus.Warnf("DisConnect rpc err :%s", err.Error())
		}
		ch.Close()
	}()

	scanner := proto.NewScanner(ch.connTcp)
	// build connect
	for scanner.Scan() {
		scanPack := new(proto.Msg)
		err := scanPack.Unpack(bytes.NewReader(scanner.Bytes()))
		if err != nil {
			logrus.Errorf("scan tcp package err:%s", err.Error())
			continue
		}
		if scanPack.Op == proto.OpBuildConn {
			opt := &proto.Connect{}
			rep := &proto.ConnectReply{}
			json.Unmarshal(scanPack.Msg, opt)
			err = logicRpc.Connect(opt, rep)
			ch.userId = rep.UserId
			ch.authToken = opt.AuthToken
			if err != nil {
				logrus.Errorf("Connect err:%s", err.Error())
				return
			} else {
				s.Bucket(ch.userId).Put(ch.userId, ch)
				break
			}
		}
	}

	heatBeatChan := make(chan bool, 1)
	defer close(heatBeatChan)

	go func() {
		for {
			select {
			case <-time.After(config.Conf.Server.PongWait):
				ch.connTcp.Close()
				return
			case _, ok := <-heatBeatChan:
				if !ok {
					logrus.Info("heatBeactChan Close")
					return
				}
			}
		}
	}()

	for scanner.Scan() {
		msg := &proto.Msg{}
		err := msg.Unpack(bytes.NewReader(scanner.Bytes()))
		if err != nil {
			logrus.Errorf("scan tcp package err:%s", err.Error())
			continue
		}

		heatBeatChan <- true
		s.Operator(msg, ch)
	}
}

func writeFromTcp(s *Server, ch *Channel) {
	defer func() {
		logrus.Info("tcpWrite close")
		ch.connTcp.Close()
	}()

	var (
		m  interface{}
		ok bool
	)

	pingTicker := time.NewTicker(config.Conf.Server.PingPeriod)

	for {
		select {
		case m, ok = <-ch.ch:
		case <-pingTicker.C:
			proto.NewMsg(proto.OpPing, nil, 0).Pack(ch.connTcp)
			continue
		}
		if !ok {
			return
		}
		switch i := m.(type) {
		case *proto.Msg:
			i.Pack(ch.connTcp)
		default:
			logrus.Warnln("connect writeFromTcp unknown type")
		}
	}
}
